package org.korosoft.rusalad.syncnrun;

import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.korosoft.rusalad.syncnrun.command.Answer;
import org.korosoft.rusalad.syncnrun.command.Command;
import org.korosoft.rusalad.syncnrun.command.FailAnswer;
import org.korosoft.rusalad.syncnrun.messaging.Message;
import org.korosoft.rusalad.syncnrun.messaging.MessageProcessor;
import org.korosoft.rusalad.syncnrun.messaging.MessageRoute;
import org.korosoft.rusalad.syncnrun.messaging.Messenger;
import org.korosoft.rusalad.syncnrun.tools.SocketFactory;
import org.korosoft.rusalad.syncnrun.tools.Utils;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Syncnrun RPC server.
 *
 * @author Dmitry Korotkov
 * @since 1.0
 */
public class Server {
    private static final Log log = LogFactory.getLog(Server.class);
    public static final int WRITE_QUEUE_CAPACITY = 128;
    private static final String paddingObject = Utils.getPaddingObject(256);
    private static final long paddingIntervalMs = 50;

    private final ExecutorService commandExecutor;
    private final SocketFactory socketFactory;

    private Thread serverThread;
    private volatile ServerSocket serverSocket;

    public Server(SocketFactory socketFactory) {
        this.socketFactory = socketFactory;
        final AtomicInteger threadNumber = new AtomicInteger(0);
        commandExecutor = Executors.newCachedThreadPool(new ThreadFactory() {
            public Thread newThread(Runnable r) {
                String name = "syncnrun-command-thread-" + threadNumber.incrementAndGet();
                if (log.isTraceEnabled()) {
                    log.trace(String.format("Spawning new thread %s", name));
                }
                return new Thread(r, name);
            }
        });
    }

    public void start(int port) throws IOException {
        log.debug(String.format("Starting server at port %d", port));
        final AtomicInteger threadNumber = new AtomicInteger(0);
        final ExecutorService executor = Executors.newCachedThreadPool(new ThreadFactory() {
            public Thread newThread(Runnable r) {
                String name = "syncnrun-thread-" + threadNumber.incrementAndGet();
                if (log.isTraceEnabled()) {
                    log.trace(String.format("Spawning new thread %s", name));
                }
                return new Thread(r, name);
            }
        });
        serverSocket = socketFactory.createServerSocket(port);

        serverThread = new Thread(new Runnable() {
            @Override
            public void run() {
                for (; ; ) {
                    final Socket client;
                    try {
                        client = serverSocket.accept();
                    } catch (Exception e) {
                        log.warn(e.getMessage());
                        log.debug(e.getMessage(), e);
                        if (serverSocket.isClosed()) {
                            break;
                        } else {
                            continue;
                        }
                    }
                    log.debug(String.format("Client connected: %s", client.getRemoteSocketAddress()));
                    try {
                        executor.execute(new ServerWorker(client));
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        }, "syncnrun-tcp-listener");
        serverThread.start();
    }

    public void stop() {
        try {
            serverSocket.close();
            await();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void await() {
        try {
            serverThread.join();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    private class ServerWorker implements Runnable, MessageProcessor {

        private final Socket socket;
        private final Messenger messenger;
        private final SynchronousStreamWriter writer;

        private ServerWorker(Socket socket) throws IOException {
            this.socket = socket;
            writer = new SynchronousStreamWriter(socket.getOutputStream());
            new Thread(writer, Thread.currentThread().getName() + "-writer").start();
            messenger = new Messenger(this);
        }

        @Override
        public void processMessage(MessageRoute route, Message message) {
            try {
                if (log.isTraceEnabled()) {
                    log.trace(String.format("Queue %s: sending message: %s", route.getName(), message));
                }
                synchronized (writer) {
                    writer.writeObject(route.getName());
                    writer.writeObject(message);
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void run() {
            final ObjectInputStream inputStream;

            try {
                inputStream = new ObjectInputStream(socket.getInputStream());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

            try {
                try {
                    for (Object o = inputStream.readObject(); o != null; o = inputStream.readObject()) {
                        if (o instanceof String) {
                            continue;
                        }
                        if (!(o instanceof UUID)) {
                            throw new RuntimeException("Command ID of class " + UUID.class.getName() + " expected. Received " + o.getClass().getName());
                        }
                        if (log.isTraceEnabled()) {
                            log.trace(String.format("Received id: %s", o));
                        }
                        final UUID commandId = (UUID) o;
                        do {
                            o = inputStream.readObject();
                        } while (o instanceof String);

                        if (o instanceof Message) {
                            if (log.isTraceEnabled()) {
                                log.trace(String.format("Queue %s: received message: %s", commandId, o));
                            }
                            try {
                                messenger.getRoute(commandId).processIncoming((Message) o);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                            continue;
                        }

                        if (!(o instanceof Command)) {
                            throw new RuntimeException("Unexpected command class: " + o.getClass().getName());
                        }
                        final Command c = (Command) o;
                        log.info(String.format("Received command: %s", c));
                        commandExecutor.execute(new Runnable() {
                            @Override
                            public void run() {
                                try {
                                    Answer a;
                                    try {
                                        a = c.execute(messenger);
                                    } catch (Throwable e) {
                                        a = new FailAnswer(e);
                                    }
                                    log.info(String.format("Sending answer: %s", a));
                                    synchronized (writer) {
                                        writer.writeObject(commandId);
                                        writer.writeObject(a);
                                    }
                                } catch (IOException e) {
                                    throw new RuntimeException(e);
                                }
                            }
                        });

                    }
                } catch (EOFException e) {
                    log.debug("Client stream ended.");
                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (ClassNotFoundException e) {
                    throw new RuntimeException(e);
                }
            } finally {
                try {
                    writer.close();
                } catch (IOException ignored) {
                }
                try {
                    inputStream.close();
                } catch (IOException ignored) {
                }
                try {
                    socket.close();
                } catch (IOException ignored) {
                }
                messenger.sendEndOfTransfer();
            }
        }
    }

    private class SynchronousStreamWriter implements Runnable, Closeable {
        private final BlockingQueue<Object> writeQueue = new ArrayBlockingQueue<Object>(WRITE_QUEUE_CAPACITY);
        private final ObjectOutputStream objectOutputStream;
        private final Semaphore semaphore = new Semaphore(1);
        private final AtomicReference<CountDownLatch> countDownLatchAtomicReference = new AtomicReference<CountDownLatch>();
        private final AtomicReference<IOException> exception = new AtomicReference<IOException>();

        private SynchronousStreamWriter(OutputStream outputStream) throws IOException {
            this.objectOutputStream = new ObjectOutputStream(outputStream);
        }

        public void writeObject(Object o) throws IOException {
            try {
                semaphore.acquire();
                try {
                    countDownLatchAtomicReference.set(new CountDownLatch(1));
                    writeQueue.put(o);
                    countDownLatchAtomicReference.get().await();
                    IOException e = exception.get();
                    if (e != null) {
                        throw e;
                    }
                } finally {
                    semaphore.release();
                }
            } catch (InterruptedException e) {
                throw new RuntimeException();
            }
        }

        public void close() throws IOException {
            try {
                writeQueue.put(this);
            } catch (InterruptedException e) {
                throw new RuntimeException();
            }
        }

        @Override
        public void run() {
            try {
                boolean lastSentObjectIsPadding = false;
                for (; ; ) {
                    final Object o;
                    if (lastSentObjectIsPadding) {
                        o = writeQueue.take();
                    } else {
                        o = writeQueue.poll(paddingIntervalMs, TimeUnit.MILLISECONDS);
                    }
                    if (o == null) {
                        if (log.isTraceEnabled()) {
                            log.trace("Sending padding object");
                        }
                        objectOutputStream.writeObject(paddingObject);
                        objectOutputStream.flush();
                        lastSentObjectIsPadding = true;
                        continue;
                    }
                    if (o == this) {
                        break;
                    }
                    try {
                        if (log.isTraceEnabled()) {
                            log.trace(String.format("Sending object %s", o));
                        }
                        objectOutputStream.writeObject(o);
                        objectOutputStream.reset();
                        lastSentObjectIsPadding = false;
                        exception.set(null);
                    } catch (IOException e) {
                        exception.set(e);
                    } finally {
                        countDownLatchAtomicReference.get().countDown();
                    }
                }
                objectOutputStream.close();
            } catch (Exception e) {
                IOUtils.closeQuietly(this);
                throw new RuntimeException(e);
            }
        }
    }
}
